package com.ruoyi.iot.handler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ruoyi.iot.config.MqttSubscribeService;
import com.ruoyi.iot.service.MqttMessageHandler;
import com.ruoyi.iotcommon.constant.DeviceConstant;
import com.ruoyi.iotcommon.constant.RocketTopicConstant;
import com.ruoyi.iotcommon.domain.OnlineStatusMessage;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Service;

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;

import java.util.Date;
// https://docs.emqx.com/zh/emqx/latest/observability/mqtt-system-topics.html#%E7%B3%BB%E7%BB%9F%E4%B8%BB%E9%A2%98
@Slf4j
@Service
@MqttSubscribeService(topic = "$SYS/brokers/+/clients/+/+")
public class OnlineStatusMqttMessageHandler implements MqttMessageHandler {

    @Resource
    private RocketMQTemplate rocketMQTemplate;


    @Override
    public void process(String topic,String message) {
        JSONObject receivedJson = JSON.parseObject(message);
        String deviceKeyAndProductKey = receivedJson.getString("username");
        String deviceKey = deviceKeyAndProductKey.split("&")[0];
        String productKey = deviceKeyAndProductKey.split("&")[1];
        //String clinetId = receivedJson.getString("clientid");
        // 构造对象
        OnlineStatusMessage onlineStatusMessage = new OnlineStatusMessage();
        onlineStatusMessage.setProductKey(productKey);
        onlineStatusMessage.setDeviceKey(deviceKey);
        onlineStatusMessage.setTime(new Date());
        if(topic.contains("connected")){
            onlineStatusMessage.setStatus(DeviceConstant.STATUS_ONLINE);
            log.info("设备上线：{}",deviceKey);
        }else{
            onlineStatusMessage.setStatus(DeviceConstant.STATUS_OFFLINE);
            log.info("设备下线：{}",deviceKey);
        }
        rocketMQTemplate.sendOneWay(RocketTopicConstant.ONLINE_STATUS, onlineStatusMessage);
    }
}
